package com.yanggu.spark.core.rdd.transform.value

import org.apache.spark.{SparkConf, SparkContext}

//Coalesce缩减分区的算子
object RDD09_Coalesce {

  def main(args: Array[String]): Unit = {
    //1. 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("mapOperator")

    //2. 创建SparkContext
    val sparkContext = new SparkContext(sparkConf)

    //3. 从内存中创建RDD
    val dataRdd = sparkContext.makeRDD(List[Int](1, 2, 3, 3 ,4, 5, 6), 6)

    //4. coalesce缩减分区
    //numPartitions 新的分区数 shuffle: Boolean 是否使用shuffle
    //4.1 这里如果有大量的小任务时(分区内的数据不多), 可以通过缩减分区数, 调用coalesce方法(传入小的分区数), 缩减分区数, 提高资源的利用率
    val value = dataRdd.coalesce(1)
    println("缩减后的分区数: " + value.getNumPartitions)

    //4.2 如果分区内数据过大, 需要使用coalesce(大的分区数, true), 提高程序的并行度
    //如果要扩大分区, 必须要进行shuffle
    val value1 = dataRdd.coalesce(7, shuffle = true)
    println("缩减后的分区数: " + value1.getNumPartitions)

    //如果没有设置shuffle为true。则分区数, 不会发生变化
    val value2 = dataRdd.coalesce(7)
    println("缩减后的分区数: " + value2.getNumPartitions)

    //5. 打印
    println(value.collect().mkString(", "))

    //6. 释放资源
    sparkContext.stop()
  }

}
